package com.doit.day02;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

//重复消费
//漏数据

/**
 * 最多一次   最多只到kafka中读取一次数据   漏读数据  读完了我马上就提交偏移量
 * 最少一次   我至少都要去kafka中读一次 ，可能读多次   读了多次就有可能处理多次   重复消费  业务处理完再提交偏移量
 * 精准一次   不多不少只读一次，并且我只处理一次
 */

public class ConsumerDemo {
    private static final String HOST = "linux01:9092,linux02:9092,linux03:9092";
    private static final String GROUP_ID = "G02";
    public static void main(String[] args) {
        Properties props = new Properties();
        /**
         * 必须得指定的
         */
//        props.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        //                                    org.apache.kafka.common.serialization.StringDeserializer
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
//        props.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
//        props.setProperty("bootstrap.servers","linux01:9092");
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,ConsumerDemo.HOST);
//        props.setProperty("group.id","g001");
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,ConsumerDemo.GROUP_ID);

        /**
         * 可选的
         */
        //允许自动创建topic
//        props.setProperty("allow.auto.create.topics","true");
        props.setProperty(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG,"true");
        //对于从哪里消费的位置指定 earliest   latest
//        props.setProperty("auto.offset.reset","earliest");
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        //运行自动提交偏移量
//        props.setProperty("enable.auto.commit","true");
        //都是你惹的祸，不要你了，分手吧
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
        //提交偏移量的间隔时间
//        props.setProperty("auto.commit.interval.ms","10000");
//        props.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"10000");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        //还要告诉他，我拉取的topic是谁，有没有可能我一次性消费多个topic中的数据呢？
/*        ArrayList<String> list = new ArrayList<>();
        list.add("java");
        list.add("test");*/
        //订阅拉取数据的主题
        consumer.subscribe(Arrays.asList("java"));
/*        TopicPartition topicPartition = new TopicPartition("java", 0);
        TopicPartition topicPartition1 = new TopicPartition("java", 1);
        TopicPartition topicPartition2 = new TopicPartition("java", 2);
        consumer.assign(Arrays.asList(topicPartition,topicPartition1,topicPartition2));
        //指定从哪里开始消费
        consumer.seek(topicPartition,3);
        consumer.seek(topicPartition1,10);
        consumer.seek(topicPartition2,20);*/

        while (true){
            //拉取数据  一次拉取一批数据   拉取数据  100条数据
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));
            //给你一个api  =》 就是用来提交偏移量的api

            //手动让他提交偏移量的api   还是会有问题滴   但是只可能会漏读数据
            //半自动提交
            consumer.commitAsync();
            //1-100
            //就是将上面拉取的100条数据做业务的处理
            for (ConsumerRecord<String, String> consumerRecord : poll) {
                //第一种方式叫去重逻辑
                //还有一种，幂等性
                String topic = consumerRecord.topic();
                int partition = consumerRecord.partition();
                long offset = consumerRecord.offset();
                String value = consumerRecord.value();
                //做具体的业务处理，或者说叫具体的数据分析
                //把数据写入到mysql中
                System.out.println("topic:"+topic+",partition:"+partition+",offset:"+offset+",value:"+value);
            }
            //程序中断了
           //手动让他提交偏移量的api   还是会有问题滴   但是只可能会重复消费数据
            consumer.commitAsync();

            //程序中断了呢

            //还没来得及提交偏移量

        }



    }
}
